package com.atguigu.flink.chapter04_window.agg;

import com.atguigu.flink.pojo.MyAcc;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2022/10/25
 *
 * Aggregate: 累加器。
 *          求每种传感器，每5个的平均vc
 *
 *                  AggregateFunction<IN, ACC, OUT>
 *                          封装累加的数据为Tuple
 *                          封装累加的数据为Bean
 *                          
 *                          
 *  aggregate(
 *             AggregateFunction<T, ACC, V> aggFunction,   //滚动聚合，来一条聚合一次
 *             WindowFunction<V, R, K(key), W> windowFunction  //窗口被触发时，执行一次
 *             )
 *
 */
public class Demo13_Aggregate
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env
           .socketTextStream("hadoop103", 8888)
           .map(new MapFunction<String, WaterSensor>()
           {
               @Override
               public WaterSensor map(String value) throws Exception {
                   String[] data = value.split(",");
                   return new WaterSensor(
                       data[0],
                       Long.valueOf(data[1]),
                       Integer.valueOf(data[2])
                   );
               }
           })
           .keyBy(WaterSensor::getId)
           .countWindow(3,2)
           .aggregate(new AggregateFunction<WaterSensor, MyAcc, Double>()
           {
               //初始化一个累加器
               @Override
               public MyAcc createAccumulator() {
                   return new MyAcc();
               }

               //累加
               @Override
               public MyAcc add(WaterSensor value, MyAcc acc) {
                   System.out.println("Demo13_Aggregate.add");
                   //个数+1
                   acc.setCount(acc.getCount() + 1);

                   acc.setSum(acc.getSum() + value.getVc());

                   return acc;
               }

               //返回最终的结果
               @Override
               public Double getResult(MyAcc acc) {

                   return acc.getSum() * 1.0d / acc.getCount();
               }

               /*
                    流处理不用实现，批处理可以去实现
                    把两个累加器，合并。
                */
               @Override
               public MyAcc merge(MyAcc a, MyAcc b) {
                   return null;
               }
           }, new WindowFunction<Double, String, String, GlobalWindow>()
           {
               // Iterable<Double> input : 只有一个元素，就是AggregateFunction的输出
               @Override
               public void apply(String key, GlobalWindow window, Iterable<Double> input, Collector<String> out) throws Exception {
                    out.collect(key + ":" +input.iterator().next());
               }
           })
           .print();

        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }


}
